<?php
namespace App\Lib\RpcClient;

use App\Lib\Consul\ServiceHelper;
use App\Lib\Consul\Service;
use App\Lib\Consul\ServiceInfo;
use App\Lib\Consul\ServiceSelector;
use Exception;
use Swoft\Bean\Annotation\Mapping\Bean;
use Swoft\Bean\Annotation\Mapping\Inject;
use Swoft\Validator\Exception\ValidatorException;

/**
 * Class RpcClient
 * @package App\Lib\RpcClient
 * @Bean()
 */
class RpcClient extends AbstractClient
{

    const RPC_EOL = "\r\n\r\n";

    public const RPC_LIB = "\\App\\Rpc\Lib\\";

    /**
     * @Inject()
     * @var RpcResult
     */
    protected $RpcResult;

    /**
     * @Inject()
     * @var ServiceSelector
     */
    protected $ServiceSelector;


    /**
     * @Inject()
     * @var ServiceHelper
     */
    protected $ServiceHelper;

    /**
     * 根据 Prefix 和 serviceName 获取RpcClient
     * @param null $servicePrefix
     * @param null $serviceName
     * @return RpcClient
     */
    public function getService ($servicePrefix = null, $serviceName = null) {
        $service = clone $this;
        $service->servicePrefix = $servicePrefix ?? $this->servicePrefix;
        $service->serviceName = $serviceName ?? $this->serviceName;
        return $service;

    }

    /**
     * 根据serviceInfo (如：Common.User.UserInterface.find) 获取RpcClient
     * @param string $serviceInfo
     * @return RpcClient
     */
    public function getServiceByServiceInfo (string $serviceInfo)
    {
        // 克隆当前类 防止Bean 注入后 出现类的相互污染
        $service = clone $this;
        // 实例化ServiceInfo 类
        $serviceInfo = new ServiceInfo($serviceInfo);
        // 如果实例化后ServiceInfo 类不可用则返回 service
        if (!$serviceInfo->useful()) {
            return $service;
        }
        // 设置Prefix
        $service->setServicePrefix( $serviceInfo->getPrefix() );
        // 设置 serviceName
        $service->setServiceName( $serviceInfo->getServiceName() );
        // 设置 服务类
        $service->setClass( $serviceInfo->getServer() );
        // 设置 服务方法
        $service->setAction( $serviceInfo->getAction() );
        return $service;
    }

    /**
     * 在Consul中获取服务
     * @return Service|bool
     * @throws \Swoft\Consul\Exception\ClientException
     * @throws \Swoft\Consul\Exception\ServerException
     */
    protected function selectService () {
        // 如果没有服务名称返回false
        if (empty($this->serviceName)) {
            return false;
        }
        //筛选服务,local走原来的consul,其他自定义
        if(env('APP_ENV', 'local') === 'local'){
            // 是否自动筛选 默认不是自动的
            $auto = false;
            // 如果没有 prefix 则返回错误
            if (empty($this->servicePrefix)) {
                $auto = true;
                $this->servicePrefix = getRpcTenant();
            }
            $services = $this->ServiceSelector->services($this->servicePrefix . '_' . $this->serviceName);
            if (empty($services) && $auto) {
                $this->servicePrefix = env("DEFAULT_SERVICE_PREFIX", 'Common');
                $services = $this->ServiceSelector->services($this->servicePrefix . '_' . $this->serviceName);
            }
            /** @var Service $service */
            $service = $this->ServiceHelper->random($services);
        }else{
            //生成oos-test
            if(env('APP_ENV', 'local') == 'v6'){
                $service['Meta']['SERVICE_ADDRESS'] = strtolower($this->serviceName).'.'.env('APP_ENV', 'local').config('service.consul_dns', '.micro-service-test.svc.cluster.local');
            }else{
                $service['Meta']['SERVICE_ADDRESS'] = strtolower($this->serviceName).'-'.env('APP_ENV', 'local').config('service.consul_dns', '.micro-service-test.svc.cluster.local');
            }
            $service['Meta']['SERVICE_PORT'] = config('service.rpc_service_port', 18307);
            $service['Meta']['LIMITER_RATE'] = config('service.limiter_rate', 10000);
            $service['Meta']['LIMITER_MAX'] = config('service.limiter_max', 10000);
            $service = new Service($service);
        }
        return $service;
    }

    /**
     * @return bool|mixed
     * @throws RpcConnectException
     * @throws \Swoft\Consul\Exception\ClientException
     * @throws \Swoft\Consul\Exception\ServerException
     */
    public function query()
    {
        // 获取服务
        $service = $this->selectService();
        if (empty($service)) {
            return false;
        }
        // 如果获取到的服务不可用则返回false
        if (!$service->useful()) {
            return false;
        }
        $result = $this->request($service->getHost(),
            $service->getNamespace(). "\\" . $this->getClass(),
             $this->getAction(), $this->params,
            $this->version, $this->ext);

        //rpc无返回值做了处理@date 2020/07/16
        return $result?$this->RpcResult->newResult($result):[];
    }


    /**
     * @param $host
     * @param $class
     * @param $method
     * @param $param
     * @param string $version
     * @param array $ext
     * @return mixed
     * @throws Exception
     */
    protected function request($host, $class, $method, $param, $version = '1.0', $ext = []) {
        $fp = stream_socket_client($host, $errno, $errstr);
        if (!$fp) {
            throw new Exception("stream_socket_client fail errno={$errno} errstr={$errstr}");
        }

        if(strtolower($this->getRpcType()) == 'hyperf'){
            /**
             * this is the hyperf standard
             */
            $req = [
                "jsonrpc" => '2.0',
                "method" => sprintf("%s", $class .'/'. $method),
                'params' => $param?[$param]:[],
                'id' => '',
                'context' => $ext,
            ];
        }else{
            /**
             * this is the swoft standard
             */
            $req = [
                "jsonrpc" => '2.0',
                "method" => sprintf("%s::%s::%s", $version, self::RPC_LIB . $class . '::class', $method),
                'params' => $param,
                'id' => '',
                'ext' => $ext,
            ];
        }

        $data = json_encode($req) . self::RPC_EOL;
        fwrite($fp, $data);

        $result = '';
        while (!feof($fp)) {
            $tmp = stream_socket_recvfrom($fp, 1024);

            if ($pos = strpos($tmp, self::RPC_EOL)) {
                $result .= substr($tmp, 0, $pos);
                break;
            } else {
                $result .= $tmp;
                if(strlen($tmp) < 1024){
                    break;
                }
            }
        }

        fclose($fp);
        return json_decode($result, true);
    }

}
